package com.test.flink_source.source;

import org.apache.flink.api.common.functions.AbstractRichFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class MyRichSource extends RichSourceFunction<String> {

    private boolean flag = true;

    // 程序运行时执行一次
    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("open------------------------");
    }

    // 程序结束时执行一次
    @Override
    public void close() throws Exception {
        System.out.println("close------------------------");
    }

    @Override
    public RuntimeContext getRuntimeContext() {
        return super.getRuntimeContext();
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (flag) {
            String res = UUID.randomUUID().toString().replace("-", "");
            ctx.collect(res);

            TimeUnit.SECONDS.sleep(2);
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }
}
